package com.framework.flink;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.aggregation.Aggregations;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collector;

/**
 * 描述：
 * 读取mysql采用JDBCInputFormat 读取
 *   基于dataset
 * @author xianggj
 * @Date 2021/10/19 11:13
 **/
public class FlinkTestMysqlInput {

    public static void main(String[] args) throws Exception {

//t1只有一列a int型

        TypeInformation[] fieldTypes =new TypeInformation[]{
                BasicTypeInfo.INT_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.INT_TYPE_INFO

        };

        RowTypeInfo rowTypeInfo =new RowTypeInfo(fieldTypes);

        JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat()

                .setDrivername("com.mysql.jdbc.Driver")

                .setDBUrl("jdbc:mysql://192.168.1.37:3306/test")

                .setUsername("root")

                .setPassword("123456")

                .setQuery("select *  from test2")

                .setRowTypeInfo(rowTypeInfo)

                .finish();

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSource s = env.createInput(jdbcInputFormat); //datasource

        BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);

        tableEnv.createTemporaryView( "test2",s);

        tableEnv.sqlQuery("select * from test2").printSchema();

        Table query = tableEnv.sqlQuery("select * from test2");

        DataSet<Row> result = tableEnv.toDataSet(query, Row.class);
        List<Row> collect1 = result.collect();
        for (Row row : collect1) {
            System.out.println(row.toString());
        }
        TypeInformation<Row> type = result.getType();
        RowTypeInfo rowTypeInfo1 = changeType(2, result);
        //转换列类型
        String callLocation = Utils.getCallLocationName();
        result = new MapOperator<Row, Row>(result, rowTypeInfo1, e ->{
                //类型不为空才进行转换
                BasicTypeInfo basicTypeInfo = BasicTypeInfo.DOUBLE_TYPE_INFO;
                Object field = e.getField(2);
                if (field != null){
                    Object covert = null;
                    //转换值类型
                    try {
                        covert = covert(field, basicTypeInfo);
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                    e.setField(2, covert);
                }
            return e;
        }, callLocation);
        List<Row> collect = result.collect();
        System.out.println(collect.size());
//        System.out.println(s.count());
    }

    private static RowTypeInfo changeType(int i, DataSet<Row> dataset) {
        RowTypeInfo type = (RowTypeInfo)dataset.getType();
        TypeInformation<?>[] fieldTypes = type.getFieldTypes();
        fieldTypes[i] = BasicTypeInfo.DOUBLE_TYPE_INFO;
        return new RowTypeInfo(fieldTypes, type.getFieldNames());
    }

    public static Object covert(Object oldVal, BasicTypeInfo typeInfo){
        if (oldVal == null) {
            return null;
        }
        String valStr = oldVal.toString();
        if (BasicTypeInfo.STRING_TYPE_INFO.equals(typeInfo)){
            return valStr;
        } else if (BasicTypeInfo.INT_TYPE_INFO.equals(typeInfo)){
            //double 四舍五入
            return (int) Math.round(Double.valueOf(valStr));
        } else if (BasicTypeInfo.DOUBLE_TYPE_INFO.equals(typeInfo)){
            return Double.valueOf(valStr);
        } else if (BasicTypeInfo.CHAR_TYPE_INFO.equals(typeInfo)){
            return valStr.charAt(0);
        } else if (BasicTypeInfo.DATE_TYPE_INFO.equals(typeInfo)){
            return DateUtil.parse(valStr);
        } else if (BasicTypeInfo.LONG_TYPE_INFO.equals(typeInfo)){
            return Long.valueOf(StrUtil.subBefore(valStr, ".", true));
        } else if (BasicTypeInfo.BIG_INT_TYPE_INFO.equals(typeInfo)){
            return new BigInteger(StrUtil.subBefore(valStr, ".", true));
        } else if (BasicTypeInfo.FLOAT_TYPE_INFO.equals(typeInfo)){
            return Float.valueOf(valStr);
        } else if (BasicTypeInfo.SHORT_TYPE_INFO.equals(typeInfo)){
            return Short.valueOf(StrUtil.subBefore(valStr, ".", true));
        } else if (BasicTypeInfo.BOOLEAN_TYPE_INFO.equals(typeInfo)){
            return Boolean.valueOf(valStr);
        } else if (BasicTypeInfo.BYTE_TYPE_INFO.equals(typeInfo)){
            return Byte.valueOf(valStr);
        } else if (BasicTypeInfo.BIG_DEC_TYPE_INFO.equals(typeInfo)){
            return new BigDecimal(valStr);
        }
        return oldVal;
    }
}
